package mn10;


import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.GroupPattern;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.Quantifier;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;
import java.util.List;
import java.util.Map;
//使用CEP检测用户是否恶意下单，规则为5分钟内连续下单并取消超过10次
public class DemoTest04 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<OrderEvent> orderEventSingleOutputStreamOperator = env.fromElements(
                new OrderEvent("1", 1000, true),
                new OrderEvent("1", 2000, false),
                new OrderEvent("1", 3000, true),
                new OrderEvent("1", 4000, false),
                new OrderEvent("1", 5000, true),
                new OrderEvent("1", 6000, false),
                new OrderEvent("1", 7000, true),
                new OrderEvent("1", 8000, false),
                new OrderEvent("1", 9000, true),
                new OrderEvent("1", 10000, false),
                new OrderEvent("1", 12000, true),
                new OrderEvent("1", 13000, false),
                new OrderEvent("1", 14000, true),
                new OrderEvent("1", 15000, false),
                new OrderEvent("1", 16000, true),
                new OrderEvent("1", 17000, false),
                new OrderEvent("1", 18000, true),
                new OrderEvent("1", 19000, false),
                new OrderEvent("1", 20000, true),
                new OrderEvent("1", 21000, false),


                new OrderEvent("2", 1000, true),
                new OrderEvent("2", 2000, false),
                new OrderEvent("2", 3000, true),
                new OrderEvent("2", 4000, false),
                new OrderEvent("2", 5000, true),
                new OrderEvent("2", 6000, false),
                new OrderEvent("2", 7000, true),
                new OrderEvent("2", 8000, false),
                new OrderEvent("2", 9000, true),
                new OrderEvent("2", 10000, false),
                new OrderEvent("2", 12000, true),
                new OrderEvent("2", 13000, false),
                new OrderEvent("2", 14000, true),
                new OrderEvent("2", 15000, false),
                new OrderEvent("2", 16000, true),
                new OrderEvent("2", 17000, false),
                new OrderEvent("2", 18000, true),
                new OrderEvent("2", 19000, false),
                new OrderEvent("2", 20000, true),
                new OrderEvent("2", 21000, false)
        ).assignTimestampsAndWatermarks(WatermarkStrategy.
                <OrderEvent>forMonotonousTimestamps().withTimestampAssigner(
                new SerializableTimestampAssigner<OrderEvent>() {
                    @Override
                    public long extractTimestamp(OrderEvent element, long recordTimestamp) {
                        return element.getTimestamp();
                    }
                }));
        KeyedStream<OrderEvent, String> keyedStream =
                orderEventSingleOutputStreamOperator.keyBy(a -> a.getUserId());

        GroupPattern<OrderEvent, OrderEvent> begin = Pattern.begin(
                Pattern.<OrderEvent>begin("one")
                        .where(new SimpleCondition<OrderEvent>() {
                            @Override
                            public boolean filter(OrderEvent value) throws Exception {
                                return value.isOrder();
                            }
                        }).next("two")
                        .where(new SimpleCondition<OrderEvent>() {
                            @Override
                            public boolean filter(OrderEvent value) throws Exception {
                                return !value.isOrder();
                            }
                        })
        );
        //consecutive()表名10次匹配必须紧挨着
        Pattern<OrderEvent, OrderEvent> pattern = begin.times(10).consecutive().within(Time.minutes(5));


        PatternStream<OrderEvent> pattern1 = CEP.pattern(keyedStream, pattern);

        SingleOutputStreamOperator<String> one = pattern1.select(new PatternSelectFunction<OrderEvent, String>() {
            @Override
            public String select(Map<String, List<OrderEvent>> map) throws Exception {
                System.out.println(map);
                OrderEvent one = map.get("one").get(0);
                return one.getUserId() + "用户连续下单10次并取消订单10次";
            }
        });
        one.print("-------->");
        env.execute();
    }
}
